package com.clsz.consumer;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.clsz.bean.bo.OrdersearchBo;
import com.clsz.bean.po.*;
import com.clsz.mapper.OrderMapper;
import com.clsz.utils.ESResponse;
import com.clsz.utils.ESUtils;
import com.clsz.utils.RedisUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.math.BigDecimal;
import java.util.Date;
import java.util.UUID;

@Component
public class OrderConsumer {
    @Resource
    private RedisUtils redisUtils;
    @Resource
    private ESUtils esUtils;
    @Resource
    private OrderMapper orderMapper;

    @KafkaListener(topics = "order",groupId = "group-01")
    public void addOrder(ConsumerRecord<String,String> record, Acknowledgment ack) {
        //1接收到的kafkamq中的order话题
        OrdersearchBo ordersearchBo = JSONObject.parseObject(record.key(),OrdersearchBo.class);
        //2写入数据库
        UserInfo userInfo = (UserInfo) redisUtils.get(record.value());
        //获取场次实体模型对象
        SearchSourceBuilder searchSourceBuilder = null;
        BoolQueryBuilder boolQueryBuilder = null;
        //根据watchtimesId获得对应的对象
        searchSourceBuilder = new SearchSourceBuilder();
        boolQueryBuilder = new BoolQueryBuilder();
        boolQueryBuilder.must(QueryBuilders.termQuery("wt_id", ordersearchBo.getWatchTimesId()));
        searchSourceBuilder.query(boolQueryBuilder);
        WatchTimes watchTimes = esUtils.search("watch_times", searchSourceBuilder, WatchTimes.class).getData().get(0);
        //根据watchtimes中的cinemaid获取影院信息
        searchSourceBuilder = new SearchSourceBuilder();
        boolQueryBuilder = new BoolQueryBuilder();
        boolQueryBuilder.must(QueryBuilders.termQuery("cma_id",watchTimes.getCmaId()));
        searchSourceBuilder.query(boolQueryBuilder);
        CinemaInfo cinema = esUtils.search("cinema", searchSourceBuilder, CinemaInfo.class).getData().get(0);
        //获取影片信息
        searchSourceBuilder = new SearchSourceBuilder();
        boolQueryBuilder = new BoolQueryBuilder();
        boolQueryBuilder.must(QueryBuilders.termQuery("film_id", watchTimes.getFilmId()));
        searchSourceBuilder.query(boolQueryBuilder);
        FilmInfo filmInfo = esUtils.search("film", searchSourceBuilder, FilmInfo.class).getData().get(0);
        //装载
        Orders orders = new Orders();
        orders.setOrderNo(UUID.randomUUID().toString());
        orders.setOrderTime(new Date());
        orders.setOrderUserId(userInfo.getUserId());
        orders.setOrderUserNick(userInfo.getUserNickName());
        orders.setOrderCinemaId(cinema.getCmaId());
        orders.setOrderCinemaName(cinema.getCmaName());
        orders.setOrderFilmId(watchTimes.getFilmId());
        orders.setOrderFilmName(filmInfo.getFilmName());
        orders.setOrderWdDate(watchTimes.getWdDate());
        orders.setOrderWtBegintime(watchTimes.getWtBegintime());
        orders.setOrderWtEndtime(watchTimes.getWtEndtime());
        orders.setOrderWtHalls(watchTimes.getWtHalls());
        orders.setOrderCost(new BigDecimal((watchTimes.getWtCost().doubleValue()) * (ordersearchBo.getSiteBoList().size())));
        //将
        orders.setOrderSites(JSON.toJSONString(ordersearchBo.getSiteBoList()));
        orders.setOrderWtId(ordersearchBo.getWatchTimesId());
        //0未同步 1同步
        orders.setOrderIsUse((byte) 0);
        orders.setOrderState((byte) 1);
        //添加到MySQL
        int affectRow = orderMapper.insert(orders);
        if (affectRow > 0) {
            //成功后回调
            ack.acknowledge();
        }
    }
}
